package com.example.flinkcdcmysql;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkSqlCdc {

    public static void main(String[] args) throws Exception {

        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2.DDL方式建表
        tableEnv.executeSql("CREATE TABLE mysql_binlog ( " +
                " id bigint NOT NULL, " +
                " name varchar(200) " +
                ") WITH ( " +
                " 'connector' = 'mysql-cdc', " + // 连接器的名称
                " 'hostname' = 'node103', " +
                " 'port' = '3306', " +
                " 'username' = 'dev_fqr', " +
                " 'password' = 'Dev@fqr2021!', " +
                " 'database-name' = 'test_jdbc1', " +
                " 'table-name' = 'test_standard_1' " +
                ")");

        // 3.查询数据
        Table table = tableEnv.sqlQuery("select * from mysql_binlog");
        // 4.将动态表转换为流
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(table, Row.class);
        tuple2DataStream.print();
        // 5.启动任务
        env.execute("Flink-SQL-CDC");

    }

}
